package com.atguigu.bigdata.streaming

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}


object SparkStreamingStudy_window1 {
  def main(args: Array[String]) {

    val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount")
    //批次为5秒
    val ssc = new StreamingContext(conf, Seconds(5))

    // Create a DStream that will connect to hostname:port, like localhost:9999
    val lines = ssc.socketTextStream("10.21.13.181", 9999)

    // Split each line into words
    val words = lines.flatMap(_.split(" "))

    // Count each word in each batch
    val pairs = words.map(word => ("Sum",word.trim.toInt))
    //窗口大小为24小时,滑动步长为5s ======》每间隔5秒都计算前一天的数据   滑动步长必须是批次的倍数
    val wordCounts = pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b),Seconds(86400), Seconds(5))

    // Print the first ten elements of each RDD generated in this DStream to the console
    wordCounts.print()
    wordCounts.foreachRDD(rdd=>{
      //rdd[(string,int)]转换为rdd[Int]
      //rddSum[Int]
      val rddSum = rdd.map{(x)=>(x._2)}
      println(rddSum.first())
    })

    ssc.start()             // Start the computation
    ssc.awaitTermination()  // Wait for the computation to terminate
    //ssc.stop()
  }

}
